package dal

import (
	"context"
	"fmt"
	"strconv"
	"time"

	"gorm.io/plugin/dbresolver"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/kv"
	"github.com/spf13/cast"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/spec"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/service/object"
	"code.byted.org/gopkg/logs/v2/log"
	"github.com/pkg/errors"
	"gorm.io/gorm"
)

// string, value = binary(v1.Element)
func genElementKey(unique spec.ObjectUnique) string {
	return fmt.Sprintf("ele:%s:%s:%s",
		unique.GetAppId(), unique.GetRoomId(), unique.GetObjId())
}

// hash, field = objectId, value = ActionCount
func genActionCountKey(unique spec.PageUnique) string {
	return fmt.Sprintf("actCnt:%s:%s:%d:%s",
		unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId())
}
func genDelObjectListKey(unique spec.PageUnique) string {
	return fmt.Sprintf("objDelList:%s:%s:%d:%s",
		unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId())
}

// CreateElement 在数据库中插入唯一的 object
func (r *repo) CreateElement(ctx context.Context, element object.Element) error {
	obj, err := ddl.NewObjects(element)
	if err != nil {
		return err
	}

	err = r.db.WithContext(ctx).Table(ddl.TableObjects).Create(obj).Error
	if err != nil {
		if IsDuplicateError(err) {
			return biz_error.Conflict.Wrap(errors.WithStack(err))
		}
		return err
	}

	r.updateObjectCache(ctx, element, obj.Data)
	return nil
}

func (r *repo) updateObjectCache(ctx context.Context, element object.Element, data []byte) {
	actionCountKey := genActionCountKey(element)
	pipe := r.cache.NewPipeline(ctx)
	pipe.Set(genElementKey(element), string(data), time.Hour)
	if element.GetInitStatus() {
		pipe.HSet(actionCountKey, element.GetObjId(), strconv.FormatInt(int64(element.GetActionCount()), 10))
		pipe.Expire(actionCountKey, time.Hour*24)
	}
	_, err := pipe.Exec()
	if err != nil {
		log.V1.CtxWarn(ctx, "%+v", err)
	}
}

// GetElement 先读 cache  再读 db
func (r *repo) GetElement(ctx context.Context, unique spec.ObjectUnique) (object.Element, error) {
	data, err := r.cache.Get(ctx, genElementKey(unique))
	if err != nil && !IsNotFoundError(err) {
		return nil, err
	}
	if len(data) != 0 {
		return object.NewElementFromRaw(unique.GetAppId(), unique.GetRoomId(), []byte(data))
	}
	var objs []ddl.Objects
	err = r.db.WithContext(ctx).Table(ddl.TableObjects).
		Where("app_id = ? and room_id = ? and object_id = ?",
			unique.GetAppId(), unique.GetRoomId(), unique.GetObjId()).
		Find(&objs).Error
	if err != nil {
		return nil, err
	}
	if len(objs) == 0 {
		log.V1.CtxWarn(ctx, "not exist target object object rows appId:%s,roomId:%s,objectId:%s", unique.GetAppId(), unique.GetRoomId(), unique.GetObjId())
		return nil, biz_error.NotFound.Wrap(errors.Errorf("not exist target object obj:%s", unique.GetObjId()))
	}

	return object.NewElementFromRaw(objs[0].AppId, objs[0].RoomId, objs[0].Data)
}

func (r *repo) SaveElement(ctx context.Context, element object.Element) error {
	obj, err := ddl.NewObjects(element)
	if err != nil {
		return err
	}
	result := r.db.WithContext(ctx).Table(ddl.TableObjects).
		Where("app_id = ? and room_id = ? and object_id = ?",
			obj.AppId, obj.RoomId, obj.ObjectId).
		Updates(ddl.Objects{Data: obj.Data, ActionCount: obj.ActionCount, Initialized: obj.Initialized})
	log.V1.CtxInfo(ctx, "update %d rows", result.RowsAffected)
	if result.Error != nil {
		return biz_error.DatabaseError.Wrap(result.Error)
	}
	if result.RowsAffected != 1 {
		return biz_error.DatabaseError.Wrap(errors.Errorf("fail to update obj %s", obj.ObjectId))
	}

	r.updateObjectCache(ctx, element, obj.Data)
	return nil
}
func genElementsKey(unique spec.PageUnique, objIds []string) string {
	var objIdsStr string
	for _, id := range objIds {
		objIdsStr += cast.ToString(id) + "-"
	}
	return fmt.Sprintf("elements:%s:%s:%d:%s:%s",
		unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId(), objIdsStr)
}

// GetElements 是否能容忍主从不一致，缓存/DB 不一致?
func (r *repo) GetElements(ctx context.Context, unique spec.PageUnique, ids []string) ([]object.Element, error) {
	ret, err, _ := MSingleflight.Do(genElementsKey(unique, ids), func() (interface{}, error) {
		var objs []ddl.Objects
		query := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TableObjects).
			Where("app_id = ? and room_id = ? and board_id = ? and page_id = ?",
				unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId())
		if len(ids) != 0 {
			query = query.Where("object_id in ?", ids)
		}
		err := query.Find(&objs).Error
		if err != nil {
			return nil, biz_error.DatabaseError.Wrap(errors.WithStack(err))
		}
		elements := make([]object.Element, 0, len(objs))
		for _, obj := range objs {
			ele, err := object.NewElementFromRaw(unique.GetAppId(), unique.GetRoomId(), obj.Data)
			if err != nil {
				return nil, err
			}
			elements = append(elements, ele)
		}
		return elements, nil
	})
	if err != nil {
		return nil, err
	}

	return ret.([]object.Element), nil
}

func (r *repo) ListElementIds(ctx context.Context, unique spec.PageUnique) ([]string, error) {
	var ids []string
	err := r.db.WithContext(ctx).Clauses(dbresolver.Write).Table(ddl.TableObjects).
		Where("app_id = ? and room_id = ? and board_id = ? and page_id = ? and deleted_at is null",
			unique.GetAppId(), unique.GetRoomId(), unique.GetBoardId(), unique.GetPageId()).
		Select("object_id").Find(&ids).Error
	return ids, biz_error.DatabaseError.Wrap(errors.WithStack(err))
}

func (r *repo) DeleteElement(ctx context.Context, unique spec.PageObjectUnique) error {
	return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
		err := tx.Table(ddl.TableObjects).
			Where(`app_id = ? and room_id = ? and object_id = ?`,
				unique.GetAppId(), unique.GetRoomId(), unique.GetObjId()).
			Delete(&ddl.Objects{}).Error
		if err != nil {
			return biz_error.DatabaseError.Wrap(errors.WithStack(err))
		}
		defer r.markDeletedObject(ctx, unique)
		_, err = r.cache.HDel(ctx, genActionCountKey(unique), unique.GetObjId())
		if err != nil {
			if IsNotFoundError(err) {
				log.V1.CtxWarn(ctx, "object [%s] has no action count in cache", unique.GetObjId())
				return nil
			}
			return biz_error.CacheError.Wrap(errors.WithStack(err))
		}
		return biz_error.CacheError.Wrap(errors.WithStack(err))
	})
}

// 标记被删除对象，用于二次缓存校验
func (r *repo) markDeletedObject(ctx context.Context, unique spec.PageObjectUnique) {
	entry := kv.ZMember{
		Member: unique.GetObjId(),
		Score:  cast.ToFloat64(time.Now().Unix()),
	}
	err := r.cache.ZAdd(ctx, genDelObjectListKey(unique), entry)
	if err != nil {
		log.V1.CtxError(ctx, "object [%s] add delObjectList fail,err:%+v ,key:%s", unique.GetObjId(), err, genDelObjectListKey(unique))
	} else {
		r.cache.Expire(ctx, genDelObjectListKey(unique), time.Hour)
	}
}

func (r *repo) LockPageAudio(ctx context.Context, unique spec.PageUnique, objID string) (err error) {
	lockKey := genLockAudioKey(unique)
	expire := time.Second * 10
	ok, err := r.cache.SetNX(ctx, lockKey, objID, expire)
	if err != nil {
		return biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	if ok {
		return nil
	}
	return biz_error.Conflict.Wrap(errors.Errorf("fail to lock key %s with value %s", lockKey, objID))
}
func (r *repo) UnLockPageAudio(ctx context.Context, unique spec.PageUnique) (err error) {
	return r.cache.Del(ctx, genLockAudioKey(unique))
}
func (r *repo) LockPageVideo(ctx context.Context, unique spec.PageUnique, objID string) (err error) {
	lockKey := genLockVideoKey(unique)
	expire := time.Second * 10
	ok, err := r.cache.SetNX(ctx, lockKey, objID, expire)
	if err != nil {
		return biz_error.CacheError.Wrap(errors.WithStack(err))
	}
	if ok {
		return nil
	}
	return biz_error.Conflict.Wrap(errors.Errorf("fail to lock key %s with value %s", lockKey, objID))
}
func (r *repo) UnLockPageVideo(ctx context.Context, unique spec.PageUnique) (err error) {
	return r.cache.Del(ctx, genLockVideoKey(unique))
}
